package com.atguigu.edu.realtime.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.TradeSourceOrderBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
/**
 * @Author zhangsan
 * @Date 2022/10/21 10:19
 * @Description //TODO
 */
public class DwsTradeSourceOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)
        // TODO 3. 从 Kafka dwd_trade_order_detail 主题读取订单明细数据
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_source_order_window";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> source = env.addSource(kafkaConsumer);

        // TODO 4.转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = source.map(JSON::parseObject);

        // TODO 5. 按照 order_detail_id 分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(r -> r.getString("id"));

        // TODO 6. 去重
        SingleOutputStreamOperator<JSONObject> processedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<JSONObject>("last_value_state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastValue = lastValueState.value();
                        if (lastValue == null) {
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                            lastValueState.update(jsonObj);
                        } else {
                            String lastRowOpTs = lastValue.getString("row_op_ts");
                            String rowOpTs = jsonObj.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOpTs, rowOpTs) <= 0) {
                                lastValueState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws IOException {
                        JSONObject lastValue = this.lastValueState.value();
                        if (lastValue != null) {
                            out.collect(lastValue);
                        }
                        lastValueState.clear();
                    }
                }
        );

        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<TradeSourceOrderBean> javaBeanStream = processedStream.map(
                jsonObj -> {
                    String sourceId = jsonObj.getString("sc");
                    String orderId = jsonObj.getString("order_id");
                    String userId = jsonObj.getString("user_id");
                    Double orderAmount = jsonObj.getDouble("final_amount");
                    Long ts = jsonObj.getLong("ts") * 1000L;

                    TradeSourceOrderBean tradeSourceOrderWindow = TradeSourceOrderBean.builder()
                            .sourceId(sourceId)
                            .userIdSet(new HashSet<String>(
                                    Collections.singleton(userId)
                            ))
                            .orderIdSet(new HashSet<String>(
                                    Collections.singleton(orderId)
                            ))
                            .orderAmount(orderAmount)
                            .ts(ts)
                            .build();
                    return tradeSourceOrderWindow;
                }
        );

        // TODO 8. 设置水位线
        SingleOutputStreamOperator<TradeSourceOrderBean> withWatermarkStream = javaBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSourceOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSourceOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSourceOrderBean javaBean, long recordTimestamp) {
                                        return javaBean.getTs();
                                    }
                                }
                        )
        );

        //TODO 9 .按照来源ID分组
        KeyedStream<TradeSourceOrderBean, String> keyedByProIdStream = withWatermarkStream.keyBy(TradeSourceOrderBean::getSourceId);



        //TODO 10. 开窗
        WindowedStream<TradeSourceOrderBean, String, TimeWindow> windowDS = keyedByProIdStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)
        ));

        //TODO 11. 聚合
        SingleOutputStreamOperator<TradeSourceOrderBean> reducedStream = windowDS.reduce(
                new ReduceFunction<TradeSourceOrderBean>() {
                    @Override
                    public TradeSourceOrderBean reduce(TradeSourceOrderBean value1, TradeSourceOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(
                                value2.getOrderIdSet()
                        );
                        value1.getUserIdSet().addAll(
                                value2.getUserIdSet()
                        );
                        value1.setOrderAmount(
                                value1.getOrderAmount() + value2.getOrderAmount()
                        );
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeSourceOrderBean, TradeSourceOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<TradeSourceOrderBean> elements, Collector<TradeSourceOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (TradeSourceOrderBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setOrderCount((long) element.getOrderIdSet().size());
                            element.setUserCount((long) element.getUserIdSet().size());
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );

        //TODO 12.关联来源信息
        SingleOutputStreamOperator<TradeSourceOrderBean> fullInfoStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<TradeSourceOrderBean>("dim_base_source") {

                    @Override
                    public void join(TradeSourceOrderBean javaBean, JSONObject jsonObj) {
                        String sourceSite = jsonObj.getString("source_site".toUpperCase());
                        //System.out.println(sourceSite);
                        javaBean.setSourceSite(sourceSite);
                    }

                    @Override
                    public String getKey(TradeSourceOrderBean javaBean) {
                        return javaBean.getSourceId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );
        fullInfoStream.print(">>>");
        // TODO 13. 写入到 OLAP 数据库
        fullInfoStream.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_source_order_window values(?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
